package momo.app.comms;

import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.Socket;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.util.Arrays;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.ListIterator;
import java.util.StringTokenizer;

import momo.app.config.AppConfig;
import momo.app.status.StatusManager;
import momo.app.structure.Message;
import momo.app.structure.Peer;
import momo.app.structure.PeerLink;
import momo.app.structure.PendingLink;
import momo.app.structure.Relationship;
import momo.app.structure.SessionController;
import momo.app.util.FileOutput;
import momo.app.util.IntByteConvert;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;


public class RcvChannel extends Thread
{
	private Log log = LogFactory.getLog(RcvChannel.class);
	private Log logStructure = LogFactory.getLog("MoMoStructureLog");
	private Log logRcved = LogFactory.getLog("MoMoMsgRcvLog");
	
	private final int readTimeout = 1 * 30 * 1000;
	
	private boolean interrupted = false;
	private Socket sock;
//	private BufferedReader in;
	private BufferedInputStream in;
	private CommsManager commsManager;
	private StatusManager statusManager;
	
	private Peer associatedPeer;
	
	private String hisIP;
	private int hisPort;
	
	protected RcvChannel() {
		
	}
	
	public RcvChannel(Socket sock) throws IOException
	{
		this.sock = sock;
		sock.setSoTimeout(readTimeout);
//		in = new BufferedReader(new InputStreamReader(sock.getInputStream()));
		in = new BufferedInputStream(sock.getInputStream());
		
		hisIP = sock.getInetAddress().getHostAddress();
		hisPort = sock.getPort();
		
		commsManager = CommsManager.getInstance();
		statusManager = StatusManager.getInstance();
		this.start();
	}
	
	public void run()
	{
		if ( log.isDebugEnabled() )
			log.debug("RcvChannel STARTS for " + hisIP+":"+hisPort);
		if ( logStructure.isDebugEnabled() )
			logStructure.debug("RcvChannel STARTS for " + hisIP+":"+hisPort);
		
		StatusManager.getInstance().isRoot();
		
		while (!interrupted)
		{
			try
			{
				int bufferLimit = sock.getReceiveBufferSize();
				ByteBuffer byteFileSize = ByteBuffer.allocate(4);
				for (int bytecount = 0; bytecount < 4; bytecount++) {
					int byteRead = in.read();
					if (byteRead == -1)
						throw new SocketException("header malformed: EOF while reading first 4 bytes");
					else
						byteFileSize.put((byte) byteRead);
				}
				byteFileSize.rewind();
				int expectedMsgSize = byteFileSize.getInt();
				int remaining = expectedMsgSize;
				ByteBuffer msg = ByteBuffer.allocate(expectedMsgSize);
				
				while ( remaining > 0 )
				{
					int chunkSize = (remaining > bufferLimit)? bufferLimit:remaining;
					
//					System.out.println("Remainig: " + remaining);
					
					byte[] chunkBuffer = new byte[chunkSize];
					int bytesRead = in.read(chunkBuffer);
					if (bytesRead == -1)
						throw new SocketException("body malformed: EOF while reading data; expecting "+remaining+" bytes more");
	
					byte[] trimmedChunk = Arrays.copyOf(chunkBuffer, bytesRead);
					
					msg.put(trimmedChunk);
					
					remaining -= bytesRead;
				}
				//Conversion based on UTF-8
				String data = new String(msg.array(), Charset.forName("UTF-8"));
				
				if ( data == null || data.equals("") )
				{
					//when unix dies, it sends null
					
					if ( log.isDebugEnabled() )
						log.debug(hisIP+":"+hisPort+": Data rcved is null");
					if ( logStructure.isDebugEnabled() )
						logStructure.debug(hisIP+":"+hisPort+": Data rcved is null");
					
					StatusManager.getInstance().isRoot();
					
					handleUnkindLeave();
					stopRcving();
					break;
				}
				
				Message message = Message.createMessage(data);
				
				if ( message != null )
				{
					StatusManager.getInstance().isRoot();
					
					if ( log.isDebugEnabled() )
						log.debug("Message Rcved. Id: " + message.getMsgId() + ". Now in Handling process. From "+hisIP+":"+hisPort+":\n"+message.toString());
					if ( logRcved.isDebugEnabled() )
						logRcved.debug("Message Rcved. Id: " + message.getMsgId() + " Now in Handling process. From "+hisIP+":"+hisPort+":\n"+message.toString());
					handle(message);
				}else
					throw new UnsupportedEncodingException();
								
			}catch (SocketTimeoutException ste)
			{
				if ( log.isDebugEnabled() )
					log.debug(hisIP+":"+hisPort+": SocketTimeoutException", ste);
			}catch(SocketException se)
			{
				//when windows dies, it creates SocketException
				if ( log.isDebugEnabled() )
					log.debug(hisIP+":"+hisPort+": SocketException", se);
				handleUnkindLeave();
				stopRcving();
			}catch (IOException ioe)
			{
				if ( log.isErrorEnabled() )
					log.error(hisIP+":"+hisPort+": IOException", ioe);
				stopRcving();
			}
		}
		
		try
		{
			in.close();
			sock.close();
		}catch(IOException ioe)
		{
			if ( log.isErrorEnabled() )
				log.error(hisIP+":"+hisPort+": IOException", ioe);
		}
		
		if ( log.isDebugEnabled() )
			log.debug("RcvChannel ENDS for " + hisIP+":"+hisPort);
		if ( logStructure.isDebugEnabled() )
			logStructure.debug("RcvChannel ENDS for " + hisIP+":"+hisPort);
	}
	
	private void handleUnkindLeave()
	{
		if ( associatedPeer != null )
		{
			if ( log.isDebugEnabled() )
				log.debug("handleUnkindLeave");
			
			String id = associatedPeer.getId();
			
			if ( log.isDebugEnabled() )
				log.debug("Leaver id: " + id);
			if ( logStructure.isDebugEnabled() )
				logStructure.debug("Leaver id: " + id);
			
			//delayed the msg that source is sending
			Delayer.getInstance().delayMsgSent();
			
			statusManager.removeAssociationOnPeer(id);
			commsManager.removeAssociationOnPeer(id, this);
		}
	}
	
	private void handle(Message msg) throws IOException
	{
		if ( SessionController.getInstance().isSameSession( msg.getSessionId() ) )
		{
			if ( msg.getForwardType().equals(Message.FORWARD_DOWN) )
			{
				if ( logRcved.isDebugEnabled() )
					logRcved.debug("Child Rcved Down MsgId: " + msg.getMsgId() + " --> From: " + hisIP+":"+hisPort);
				if ( log.isDebugEnabled() )
					log.debug("Child Rcved Down MsgId: " + msg.getMsgId() + " --> From: " + hisIP+":"+hisPort);
				
				rcvRemoteMsg(msg);
				commsManager.forwardDown(msg);
			}else if ( msg.getForwardType().equals(Message.FORWARD_UP) )
			{
				if ( statusManager.isRoot() )
				{
					if ( logRcved.isDebugEnabled() )
						logRcved.debug("Root Rcved Up MsgId: " + msg.getMsgId() + " --> From: " + hisIP+":"+hisPort);
					if ( log.isDebugEnabled() )
						log.debug("Root Rcved Up MsgId: " + msg.getMsgId() + " --> From: " + hisIP+":"+hisPort);
					
					msg.setForwardType(Message.FORWARD_DOWN);
					rcvRemoteMsg(msg);
					commsManager.forwardDown(msg);
				}else
				{
					if ( log.isDebugEnabled() )
						log.debug("Parent Rcved Up MsgId: " + msg.getMsgId() + " --> From: " + hisIP+":"+hisPort);
					if ( logRcved.isDebugEnabled() )
						logRcved.debug("Parent Rcved Up MsgId: " + msg.getMsgId() + " --> From: " + hisIP+":"+hisPort);
					
					commsManager.forwardUp(msg);
				}
			}else if ( msg.getForwardType().equals(Message.NO_FORWARD) )
			{
				if ( msg.getMsgType().equals(Message.LINKUP) )
					replyLinkup(msg);
				else if ( msg.getMsgType().equals(Message.LINKUP_ACK) )
					replyAckLinkup(msg);
				else if ( msg.getMsgType().equals(Message.CLOSE_CONNECTION) )
					replyCloseConnection(msg);
				else if ( msg.getMsgType().equals(Message.NEW_PARENT) )
					replyNewParent(msg);
				else if ( msg.getMsgType().equals(Message.NEW_PARENT_ACK) )
					replyAckNewParent(msg);
				else if ( msg.getMsgType().equals(Message.MESSAGE_REQUEST) )
					replyMessageReq(msg);
				else if ( msg.getMsgType().equals(Message.MESSAGE_REPLY) )
					replyMessageReply(msg);
			}
		}else
		{
			int mySID = SessionController.getInstance().getSessionId();
			int msgSID = msg.getSessionId();
			if ( log.isDebugEnabled() )
				log.debug(">>>>>>>>>> Diff Session! Mine: " + mySID +", Msg: " + msgSID + " <<<<<<<<<<<<<<<<");
		}
	}
	
	private void rcvRemoteMsg(Message msg)
	{
		if ( logRcved.isDebugEnabled() )
			logRcved.debug("rcvRemoteMsg MsgId: " + msg.getMsgId());
		if ( log.isDebugEnabled() )
			log.debug("rcvRemoteMsg MsgId: " + msg.getMsgId());
		
		if ( !msg.getStates().trim().equals("") )
		{ // vector clocked message
			if ( logRcved.isDebugEnabled() )
				logRcved.debug("rcvRemoteMsg vector clock msg: before handle MsgId: " + msg.getMsgId());
			if ( log.isDebugEnabled() )
				log.debug("rcvRemoteMsg vector clock msg: before handle MsgId: " + msg.getMsgId());
			
			handleRemoteMsg(msg);
			return;
		}
		// if there is no states, it means no synch is required.
		if ( logRcved.isDebugEnabled() )
			logRcved.debug("rcvRemoteMsg no vector clock msg MsgId: " + msg.getMsgId());
		if ( log.isDebugEnabled() )
			log.debug("rcvRemoteMsg no vector clock msg MsgId: " + msg.getMsgId());			
		
		
		if ( !msg.getSourceId().equals(AppConfig.getInstance().getId()) )
		{
			if ( logRcved.isDebugEnabled() )
				logRcved.debug("rcvRemoteMsg not mine msg MsgId: " + msg.getMsgId());
			if ( log.isDebugEnabled() )
				log.debug("rcvRemoteMsg not mine msg MsgId: " + msg.getMsgId());			
						
			if ( msg.getMsgType().equals(Message.USER_DATA) )
			{
				if ( logRcved.isDebugEnabled() )
					logRcved.debug("rcvRemoteMsg user data msg MsgId: " + msg.getMsgId());
				if ( log.isDebugEnabled() )
					log.debug("rcvRemoteMsg user data msg MsgId: " + msg.getMsgId());
				
				commsManager.rcvMsg(msg);
				return;
			}
			if ( msg.getMsgType().equals(Message.UPDATE_PEER_LINK) )
			{
				if ( logRcved.isDebugEnabled() )
					logRcved.debug("rcvRemoteMsg UPDATE_PEER_LINK msg MsgId: " + msg.getMsgId());
				if ( log.isDebugEnabled() )
					log.debug("rcvRemoteMsg user UPDATE_PEER_LINK msg MsgId: " + msg.getMsgId());
				
				replyUpdatePeerLink(msg);
				return;
			}
			if ( log.isErrorEnabled() )
				logRcved.error("unknown message type for message: "+msg.toString());
			if ( log.isErrorEnabled() )
				log.error("unknown message type for message: "+msg.toString());
			
			return;
		}
		
		if ( logRcved.isDebugEnabled() )
			logRcved.debug("rcvRemoteMsg mine msg MsgId: " + msg.getMsgId());
		if ( log.isDebugEnabled() )
			log.debug("rcvRemoteMsg mine msg MsgId: " + msg.getMsgId());
		
		if ( msg.getMsgType().equals(Message.PING) )
		{
			if ( logRcved.isDebugEnabled() )
				logRcved.debug("rcvRemoteMsg PING msg MsgId: " + msg.getMsgId());
			if ( log.isDebugEnabled() )
				log.debug("rcvRemoteMsg PING msg MsgId: " + msg.getMsgId());
			
			commsManager.setSafeJoin();
			return;
		}
		if ( logRcved.isErrorEnabled() )
			logRcved.error("unknown message type for message: "+msg.toString());
		if ( log.isDebugEnabled() )
			log.error("unknown message type for message: "+msg.toString());
		return;
		
	}
	
	private void handleRemoteMsg(Message msg)
	{
		List<Message> listMsg = CommsHistory.getInstance().rcvMsg(msg);
		Iterator<Message> iter = listMsg.iterator();
		
		while (iter.hasNext() )
		{
			Message message = iter.next();
			if ( message.getMsgType().equals(Message.USER_DATA) )
				commsManager.rcvMsg(message);
			else {
				if ( logRcved.isErrorEnabled() )
					logRcved.error("Unknown message type for message: "+message.toString());
			}
			//old way of doing. Moving update peer link as no vector clock msg
//			if ( message.getMsgType().equals(Message.UPDATE_PEER_LINK) )
//				replyUpdatePeerLink(message);
//			else if ( message.getMsgType().equals(Message.USER_DATA) )
//				commsManager.rcvMsg(message);
		}
	}
	
	public boolean isSameAssociatedPeer(Peer peer)
	{
		if ( associatedPeer == null || peer == null )
			return false;
		
		if ( associatedPeer.getId().equals(peer.getId()) )
			return true;
		
		return false;
	}
	
	private void replyLinkup(Message msg) throws IOException
	{
		if ( log.isDebugEnabled() )
			log.debug(hisIP+":"+hisPort+": Reply Linkup, MsgId: " + msg.getMsgId());
		
		PendingLink pend = foundPendingLink(msg.getMsg());
		if ( pend != null )
		{
			associatedPeer = pend.getPeer();
			if ( log.isDebugEnabled() )
				log.debug(hisIP+":"+hisPort+": Founding Pending Child: " + associatedPeer.getId());
			commsManager.ackLinkup(this, pend);
		}else
		{
			if ( log.isDebugEnabled() )
				log.debug(hisIP+":"+hisPort+": Fail to find Pending Child");
			stopRcving();
		}
	}
	
	private void replyAckLinkup(Message msg) throws IOException
	{
		if ( log.isDebugEnabled() )
			log.debug(hisIP+":"+hisPort+": Reply AckLinkup, MsgId: " + msg.getMsgId());
		
		Peer parent = Peer.createPeer(msg.getMsg());
		PendingLink pendLink = commsManager.removePendingPeer(parent.getId());
		if ( pendLink != null )
		{
			associatedPeer = pendLink.getPeer();
			associatedPeer.setWorkloadRate( parent.getWorkloadRate() );
			boolean success = commsManager.addParent(associatedPeer, this, pendLink.getSendChannel());
			if ( success )
			{
				statusManager.addPeer(associatedPeer, "replyAckLinkup");
				statusManager.addPeerLink(pendLink.getPeerLink());
				statusManager.getStabMapInfo(associatedPeer.getAddressToUse(), associatedPeer.getStatusPort());
				statusManager.getPeerInfo(associatedPeer.getAddressToUse(), associatedPeer.getStatusPort());
				statusManager.measureLatency();
			}
			if ( log.isDebugEnabled() )
				log.debug("Linkup Add Parent: Success?? - " + success);
		}else
		{
			if ( log.isDebugEnabled() )
				log.debug("Couldnt found pending parent. Stop rcving then");
			stopRcving();
		}
	}
	
	private PendingLink foundPendingLink(String data)
	{
		Peer p = Peer.createPeer(data);
		
		int tries = 0;
		while (tries < 5)
		{
			PendingLink pend = commsManager.removePendingPeer(p.getId());
			if ( pend != null )
				return pend;
			
			tries ++;
			
			try
			{
				Thread.sleep(5 * 1000);
			}catch (InterruptedException ie) {
				if ( log.isErrorEnabled() )
					log.error(hisIP+":"+hisPort+": InterruptedException", ie);
			}
		}
		return null;
	}
	
	private void replyUpdatePeerLink(Message msg)
	{
		if ( log.isDebugEnabled() )
			log.debug(hisIP+":"+hisPort+": UPDATE_PEER_LINK Msg id: "+msg.getMsgId());
		if ( logRcved.isDebugEnabled() )
			logRcved.debug(hisIP+":"+hisPort+": UPDATE_PEER_LINK Msg id: "+msg.getMsgId());
		
		StringTokenizer tokens = new StringTokenizer(msg.getMsg(), ",");
		
		String source_id = tokens.nextToken();
		String dest_id = tokens.nextToken();		
		double cost = Double.parseDouble(tokens.nextToken());
		double lat = Double.parseDouble(tokens.nextToken());
		
		if ( log.isDebugEnabled() )
			log.debug(" UPDATE_PEER_LINK Msg id: "+msg.getMsgId()+" lat: "+lat);
		if ( logRcved.isDebugEnabled() )
			logRcved.debug(" UPDATE_PEER_LINK Msg id: "+msg.getMsgId()+" lat: "+lat);

		if (lat == 0) {
			if ( logRcved.isErrorEnabled() )
				logRcved.error("UPDATE_PEER_LINK latency=0 for msg id:"+msg.getMsgId()+" message: "+msg);
			if ( log.isErrorEnabled() )
				log.error("UPDATE_PEER_LINK latency=0 for msg id:"+msg.getMsgId()+" message: "+msg);
		}
		
//		if lat < 0 remove the link and latMeasurement from graph else add/update
		if ( lat < 0 )
		{
			statusManager.removePeerLink(source_id, dest_id);
			statusManager.removeLatMeasureThread(associatedPeer.getId());
		}else
			statusManager.addPeerLink(source_id, dest_id, cost, lat);
	}
	
	private void replyCloseConnection(Message msg)
	{		
		Peer peerToClose = Peer.createPeer(msg.getMsg());
		
		if ( log.isDebugEnabled() )
			log.debug(hisIP+":"+hisPort+": Close Connection: " + peerToClose.getId() +", MsgId: " + msg.getMsgId());
		
		stopRcving();
		
		Relationship rs = commsManager.remove(peerToClose.getId(), this);
		boolean success = rs==null?false:true;
		
		if ( log.isDebugEnabled() )
			log.debug("replyCloseConnection Closing Peer: " + peerToClose.getId() + " - sucess: " + success);
		if ( logStructure.isDebugEnabled() )
			logStructure.debug("replyCloseConnection Closing Peer: " + peerToClose.getId() + " - sucess: " + success);
	}
	
	private void replyNewParent(Message msg) throws IOException
	{
		if ( log.isDebugEnabled() )
			log.debug(hisIP+":"+hisPort+": Reply New Parent, MsgId: " + msg.getMsgId());
		
		Peer p = Peer.createPeer(msg.getMsg());
		associatedPeer = p;
		commsManager.ackJoinNewParent(this, p);
	}
	
	private void replyAckNewParent(Message msg)
	{
		if ( log.isDebugEnabled() )
			log.debug(hisIP+":"+hisPort+": Reply AckNewParent, MsgId: " + msg.getMsgId());
		
		Peer parent = Peer.createPeer(msg.getMsg());
		PendingLink pendLink = commsManager.removePendingPeer(parent.getId());
		if ( pendLink != null )
		{
			associatedPeer = pendLink.getPeer();
			boolean success = commsManager.addParent(associatedPeer, this, pendLink.getSendChannel());
			if ( log.isDebugEnabled() )
				log.debug("ackNewParent Add Parent: Success?? - " + success);
		}else
		{
			if ( log.isDebugEnabled() )
				log.debug("Couldnt found pending parent. Stop rcving then");
			stopRcving();
		}
	}
	
	private void replyMessageReq(Message msg)
	{
		if ( log.isDebugEnabled() )
			log.debug("Rcved: Message_Request, MsgId: "+msg.getMsgId()+" ,Source: " + msg.getSourceId() +", Contents: " + msg.getMsg());
		
		StringTokenizer tokens = new StringTokenizer(msg.getMsg(), "~");
		String nodeId = tokens.nextToken();
		String msgReqSeqNos = tokens.nextToken();
		
		List<Integer> missingIds = new LinkedList<Integer>();
		StringTokenizer commaTokens = new StringTokenizer(msgReqSeqNos, ",");
		while ( commaTokens.hasMoreTokens() ) {
			String data = commaTokens.nextToken();
			int hyphenPos = data.indexOf("-");
			if ( hyphenPos < 0 )
			{
				int msgLocalSeqNo = Integer.parseInt(data);
				missingIds.add(msgLocalSeqNo);
			}
			else 
			{
				int start = Integer.parseInt(data.substring(0, hyphenPos));
				int end = Integer.parseInt(data.substring(hyphenPos+1));
				
				for(int i=start; i<=end; i++)
				{
					missingIds.add(i);
				}
			}
		}
		
		ListIterator<Integer> idsIter = missingIds.listIterator();
		while ( idsIter.hasNext() ) {
			int reqId = idsIter.next();
			
			if ( log.isDebugEnabled() )
				log.debug("missing message id: "+reqId);
			
			Message m = CommsHistory.getInstance().getMessage(nodeId, reqId);
			if ( m == null ) {
				if ( log.isDebugEnabled() )
					log.debug("requested message id: "+reqId+ " not in cache");
			} 
			else {
				if ( log.isDebugEnabled() )
					log.debug("requested message id: "+reqId+ " found in local cache");
				idsIter.remove();
				Message msgToRet = new Message(Message.NO_FORWARD, AppConfig.getInstance().getId(), Message.MESSAGE_REPLY, msg.getSourceId()+","+m.toString());
				if ( statusManager.isRoot(msg.getSourceId()) ) {
					commsManager.forwardUp(msgToRet);
				} else {
					commsManager.forwardDown(msgToRet);
				}
				if ( log.isDebugEnabled() )
					log.debug("requested message id: "+reqId+ " found and forwarded");
			}
		}
		
		if (missingIds.size() == 0) {
			if ( log.isDebugEnabled() )
				log.debug("all missing ids sent");
			return;	
		}
		
		StringBuffer missingIdsBuf = convertIdsToPrintFormat(missingIds);
		
		
		missingIdsBuf.insert(0, nodeId+"~");
		if ( log.isDebugEnabled() )
			log.debug("still missing ids: "+missingIdsBuf);
		
		msg.setMsg(missingIdsBuf.toString());
		if ( statusManager.isRoot(msg.getSourceId()) ) {
			commsManager.forwardDown(msg);
		} else {
			commsManager.forwardUp(msg);
		}
		if ( log.isDebugEnabled() )
			log.debug("message with still missing ids forwarded");
		
	}

	protected StringBuffer convertIdsToPrintFormat(List<Integer> missingIds) {
		StringBuffer missingIdsBuf = new StringBuffer();
		int startSeries = -1;
		int lastSeen = -1;
		for (int stillMissing : missingIds) {
			if ((stillMissing - lastSeen) > 1) {
				if (startSeries != -1) {
					if (lastSeen - startSeries > 0) {
						missingIdsBuf.append(startSeries);
						missingIdsBuf.append("-");
						missingIdsBuf.append(lastSeen);
					}
					else 
						missingIdsBuf.append(startSeries);
					missingIdsBuf.append(",");
				}
				startSeries = stillMissing;
			}
			lastSeen = stillMissing;
		}
		if (lastSeen - startSeries > 0) {
			missingIdsBuf.append(startSeries);
			missingIdsBuf.append("-");
			missingIdsBuf.append(lastSeen);
		}
		else 
			missingIdsBuf.append(startSeries);
		return missingIdsBuf;
	}
	
	private void replyMessageReply(Message msg)
	{
		String contents = msg.getMsg();
		int index = contents.indexOf(",");
		String sourceId = contents.substring(0, index);
		Message m = Message.createMessage(contents.substring(index+1));
		
		if ( log.isDebugEnabled() )
			log.debug("\nRcved Message Reply (MsgId: "+msg.getMsgId()+")from source: " + msg.getSourceId() + "\nTo: " + sourceId + "\nWanted Missing Msg: " + m.getLocalSeqNo());
		
		if ( !sourceId.equals(AppConfig.getInstance().getId()) )
		{
			if ( statusManager.isRoot(sourceId) )
				commsManager.forwardUp(msg);
			else
				commsManager.forwardDown(msg);
		}
		
		handleRemoteMsg(m);
		
	}
	
	public void stopRcving()
	{
		interrupted = true;
		associatedPeer = null;
	}
	
	public void shutdownNow()
	{
		stopRcving();
		try
		{
			sock.close();
		}catch(IOException ioe)
		{
			if ( log.isErrorEnabled() )
				log.error("IOException", ioe);
		}
	}
	
	public int getSenderPort()
	{
		return hisPort;
	}
	
}//end of class RcvCommsChannel
